package cn.adley.hdfs.client.service;

import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.nio.charset.StandardCharsets;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

@Service
public class HDFSApi {
    @Value("${hdfs.namenode.name}")
    private String nameNode;
    @Value("${hdfs.address}")
    private String hdfsAddr;

    public void writeFileToHDFS(String path, String fileName, String content) throws IOException {
        Configuration configuration = new Configuration();
        configuration.set(nameNode, hdfsAddr);
        FileSystem fileSystem = FileSystem.get(configuration);

        Path hdfsWritePath = new Path(path + fileName);
        FSDataOutputStream fsDataOutputStream = fileSystem.create(hdfsWritePath,true);

        BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(fsDataOutputStream,StandardCharsets.UTF_8));
        bufferedWriter.write(content);
        bufferedWriter.newLine();
        bufferedWriter.close();
        fileSystem.close();
    }

    public void appendToHDFSFile(String path, String fileName, String content) throws IOException {
        Configuration configuration = new Configuration();
        configuration.set(nameNode, hdfsAddr);
        FileSystem fileSystem = FileSystem.get(configuration);

        Path hdfsWritePath = new Path(path + fileName);

        if (fileSystem.exists(hdfsWritePath)) {
            String originContent = this.readFile(fileSystem, hdfsWritePath);

            FSDataOutputStream fsDataOutputStream = fileSystem.create(hdfsWritePath);
            BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(fsDataOutputStream,StandardCharsets.UTF_8));
            bufferedWriter.write(originContent);
            bufferedWriter.write(content);
            bufferedWriter.newLine();
            bufferedWriter.close();
            fileSystem.close();
        } else {
            writeFileToHDFS(path,fileName,content);
        }
    }

    public String readFileFromHDFS(String path, String fileName) throws IOException {
        Configuration configuration = new Configuration();
        configuration.set(nameNode, hdfsAddr);
        FileSystem fileSystem = FileSystem.get(configuration);

        Path hdfsReadPath = new Path(path + fileName);

        String out= readFile(fileSystem, hdfsReadPath);

        fileSystem.close();
        return out;
    }

    private String readFile(FileSystem fileSystem, Path hdfsReadPath) throws IOException {
        FSDataInputStream inputStream = fileSystem.open(hdfsReadPath);
        String out= IOUtils.toString(inputStream, "UTF-8");

        inputStream.close();
        return out;
    }
}
